package DianShang_2024.ds_01.tjxt

import org.apache.hudi.org.apache.jetty.websocket.common.frames.DataFrame
import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.feature.{LabeledPoint, Normalizer, VectorAssembler}
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.functions.{avg, col, desc, expr, sum, when}
import org.apache.spark.sql.types.DoubleType
import org.apache.spark.ml.linalg.{DenseVector, Vector}

import java.util.Properties

object tjxt01 {
  def main(args: Array[String]): Unit = {
      val spark=SparkSession.builder()
        .master("local[*]")
        .appName("第一套卷子的推荐系统第一题")
        .config("hive.exec.dynamic.partition.mode","nonstrict")
//        .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
//        .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
        .enableHiveSupport()
        .getOrCreate()

    val connect=new Properties()
    connect.setProperty("user","root")
    connect.setProperty("password","123456")
    connect.setProperty("driver","com.mysql.jdbc.Driver")

    //  准备数据源
    val order_detail=spark.read.jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false","order_detail",connect)
    val order_info=spark.read.jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false","order_info",connect)
    val sku_info=spark.table("xiaoqiu.sku_info_vector")


    //  找出每个用户所购买的商品
    val user_buy_sku=order_info.join(order_detail,order_detail("order_id")===order_info("id"))
      .select("user_id","sku_id")
      //  题目要求不考虑一个商品多次购买的情况所以需要去重
      .distinct()

    //  为了解决一些从dataframe转化为dataset时候的一些数据类型找不到的问题，导入spark的隐式转换
    import spark.implicits._

    //  找出6708用户购买过的商品
    val user_6708_sku:Array[Double]=user_buy_sku
      .filter(col("user_id")===3331)
      .select("sku_id")
      .map(row => row(0).toString.toDouble)
      .collect()

    println("用户6708所购买的商品的id")
println(user_6708_sku.mkString(","))


    //  找出与用户6708所购买的相同商品最多的前10位用户
    val other_10_user_ids=user_buy_sku
      .filter(col("user_id")!==3331)
      .withColumn(
        "is_cos",
        when(col("sku_id").cast(DoubleType).isin(user_6708_sku:_*),1).otherwise(0)
      )
      .groupBy(col("user_id"))
      .agg(sum("is_cos").as("count_cos"))
      .orderBy(desc("count_cos"))
      .select("user_id")
      .map(r => r(0).toString.toLong)
      .limit(18)
      .collect()
println("和用户6708购买的商品种类最相同的前10位用户id")
  println(other_10_user_ids.mkString(","))

    //  找出相似度最高的前10位用户所购买的商品
    val other_10_sku_ids=user_buy_sku
      .filter(col("user_id").isin(other_10_user_ids:_*))
      .select("sku_id")
      .map(r => r(0).toString.toDouble)
      .collect


println("找出相似度最高的前10位用户所购买的商品")
    println(other_10_sku_ids.mkString(","))

    //  创建特征向量，方便后面的数据标准化
    //  往上面看可以看到sku_info是特征工程处理的一个中间表，对原始的sku_info做的操作就是使用了dense_rank排序
    //  sku_info.columns 返回一个包含 DataFrame sku_info 所有列名的数组或列表
    //  .tail：是返回除了第一个所有的,如果是tail(2),则表示最后两个
    val vectorAssembler=new VectorAssembler()
      .setInputCols(sku_info.columns.tail)
      .setOutputCol("features")


    val dataFrame=new Pipeline()
      .setStages(Array(vectorAssembler))
      .fit(sku_info)
      .transform(sku_info)

//  LabeledPoint 是 MLlib 中用于表示带标签的特征点的类。  labeled:标签   point:点
    //  然后对于dataframe的数据类型使用map的话就是针对每一行，而不是单独的列
    /*
        LabeledPoint 是一个用于机器学习的类，它包含两个部分：标签和特征。
    r(0).toString.toInt：提取当前行的第一列（id），将其转换为字符串后再转换为整数，作为标签。
    r(1).asInstanceOf[org.apache.spark.ml.linalg.Vector]：提取当前行的第二列（features），并将其类型转换
    为 Spark 的 Vector 类型，作为特征。
     */
    val mapData=dataFrame
      .select("id","features")
      .map(
        r => {
          LabeledPoint(r(0).toString.toInt,r(1).asInstanceOf[org.apache.spark.ml.linalg.Vector])
        }
      )

      /*
            Normalizer：这是用于标准化向量的类，可以按给定的范数（L1、L2等）对向量进行缩放。
      setInputCol("features") 指定输入列为 features。
      setOutputCol("norm_features") 指定输出列的名称为 norm_features。
      setP(2.0) 表示使用 L2 范数进行标准化。

      normalizer:标准化者
       */
    val normalizer=new Normalizer()
      .setInputCol("features")
      .setOutputCol("norm_features")
      .setP(2.0)

    val normalized_data=normalizer.transform(mapData)
      .select("label","norm_features")

    //  注册余弦相似度的udf函数
    /*
        breeze.linalg.functions.cosineDistance(...)
    使用 Breeze 库的 cosineDistance 函数计算两个向量之间的余弦距离。余弦距离是基于向量之间的夹角来度量的。
    breeze.linalg.DenseVector(v1.values)
    将 v1 的值转换为 Breeze 的 DenseVector 类型。v1.values 返回向量的数值数组。
    1 - ...
    计算余弦相似度时，余弦相似度可以表示为 1 - 余弦距离。因此，这里的代码返回的是两个向量的余弦相似度。
     */
    spark.udf.register(
      "cos",
      (v1: DenseVector, v2: DenseVector) => {
        1 - breeze.linalg.functions.cosineDistance(breeze.linalg.DenseVector(v1.values), breeze.linalg.DenseVector(v2.values))
      }
    )
println("查看标准化之后的结果")
    normalized_data.show


  //  将数据进行条件自连接并根据余弦相似度算出最高的五个商品输出
    val result=normalized_data.crossJoin(normalized_data)
      .toDF("left_label","left_norm_vector","right_label","right_norm_vector")
      .filter(col("left_label")!==col("right_label"))
      .withColumn("cos",expr("cos(left_norm_vector,right_norm_vector)"))
      .orderBy(desc("cos"))
    //  left_label用户购买时    right_label其他用户购买的并剔除指定用户购买
    .filter(col("left_label").isin(user_6708_sku:_*))
      .filter(!col("right_label").isin(user_6708_sku:_*)&& col("right_label").isin(other_10_user_ids:_*)) //完成了剔除
      .groupBy("right_label")
      .agg(avg("cos").as("cos"))
      .orderBy(desc("cos"))
      .limit(18)

    result.show

    println("---------------推荐top5结果如下-------------------")
    result.collect().zipWithIndex.foreach{
      case (row,index) =>
        val right_label=row.getAs[Double]("right_label").toInt
        val cos=row.getAs[Double]("cos")
        val output=s"相似度${index+1}(商品id:${right_label},平均相似度:${cos}"
        println(output)
    }




    spark.close()
  }

}
